package com.gugee.ins.data.collector.bloggerfollowing.listener;

import com.gugee.ins.data.collector.bloggerfollowing.config.InsBloggerFollowingConfig;
import com.gugee.ins.data.collector.bloggerfollowing.service.InsBloggerFollowingService;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;


/**
 * InsBloggerFollowingListener
 *
 * @author Huangxin
 */
@Component
@Slf4j
public class InsBloggerFollowingListener {

    /**
     * InsBloggerFollowingConfig
     */
    @Autowired
    private InsBloggerFollowingConfig insBloggerFollowingConfig;

    /**
     * insBloggerFollowerService
     */
    @Autowired
    private InsBloggerFollowingService insBloggerFollowingService;

    private InsBloggerFollowingHandler collectHandler = new InsBloggerFollowingHandler();

    /**
     * 监听任务,完成活跃粉丝的关注列表采集
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insBloggerFollowing.id}",
            topics = {"${kafka.topics.insBloggerFollowing.online}","${kafka.topics.insBloggerFollowing.priority}"},
            containerFactory = "factoryInsBloggerFollowingAck")
    public void bloggerMainCollect(List<InsBloggerMessage> messages, Acknowledgment ack) {
        collectHandler.handle("InsBloggerFollowing",messages,insBloggerFollowingConfig.getAcceptInsBloggerFollowingMsgTimeMinRange(),ack);
    }

    class InsBloggerFollowingHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insBloggerFollowingService.handleActiveFollowerFollowingCollect(messages);
        }
    }
}
